package org.jetlinks.pro.device.message;

import lombok.AllArgsConstructor;
import lombok.Generated;
import org.hswebframework.web.id.IDGenerator;
import org.jetlinks.core.event.EventBus;
import org.jetlinks.core.event.Subscription;
import org.jetlinks.core.event.TopicPayload;
import org.jetlinks.core.message.DeviceMessage;
import org.jetlinks.core.utils.TopicUtils;
import org.jetlinks.pro.PropertyConstants;
import org.jetlinks.pro.assets.AssetsHolder;
import org.jetlinks.pro.gateway.external.Message;
import org.jetlinks.pro.gateway.external.SubscribeRequest;
import org.jetlinks.pro.gateway.external.SubscriptionProvider;
import org.jetlinks.pro.utils.ReactorUtils;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;

import java.time.Duration;

@Component
@AllArgsConstructor
public class DeviceMessageSubscriptionProvider implements SubscriptionProvider {

    private final EventBus eventBus;

    @Override
    @Generated
    public String id() {
        return "device-message-subscriber";
    }

    @Override
    @Generated
    public String name() {
        return "订阅设备消息";
    }

    @Override
    @Generated
    public String[] getTopicPattern() {
        return new String[]{
            "/device/*/*/**"
        };
    }

    @Override
    public Flux<Message> subscribe(SubscribeRequest request) {
        return Flux
            .fromIterable(TopicUtils.expand(request.getTopic()))
            .flatMap(topic -> AssetsHolder
                .refactorTopic(AssetsHolder.fromAuth(request.getAuthentication()), topic)
                .flatMap(_topic -> this
                             .subscribe(request.getId(),
                                        _topic,
                                        request.getAuthentication().getUser().getId(),
                                        request.isShared())
                    , Integer.MAX_VALUE
                ))
            .map(topicMessage -> Message.success(request.getId(), topicMessage.getTopic(), topicMessage.decode(DeviceMessage.class)))
            .as(ReactorUtils
                    .distinct(msg -> {
                        DeviceMessage deviceMessage = ((DeviceMessage) msg.getPayload());
                        return deviceMessage
                            .getHeader(PropertyConstants.uid)
                            .orElseGet(IDGenerator.SNOW_FLAKE_HEX::generate);
                    }, Duration.ofSeconds(30)));
    }

    static final Subscription.Feature[] sharedFeature = {
        Subscription.Feature.local,
        Subscription.Feature.broker,
        Subscription.Feature.shared
    };

    static final Subscription.Feature[] defaultFeature = {
        Subscription.Feature.local,
        Subscription.Feature.broker
    };

    private Flux<TopicPayload> subscribe(String id, String topic, String userId, boolean shared) {
        Subscription.Feature[] features = shared ? sharedFeature : defaultFeature;

        return eventBus
            .subscribe(
                Subscription.of(
                    "DeviceMessageSubscriptionProvider:" + userId + id,
                    new String[]{topic},
                    features
                ));
    }
}
